package com.cxz.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * kafka消费者
 * @author ocxz1
 */
public class KafkaConsumerTest {

    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerTest.class);

    /**
     * 获取kafka消费端
     * @return 返回kafka消费端
     */
    public static KafkaConsumer<String, String> getConsumer() {

        // 根据默认Kafka消费者配置，创建
        return  new KafkaConsumer<>(KafkaCommonProperties.getDefaultKafkaConsumerConfig());
    }


    public static void main(String[] args) {

        // 获取消费端
        KafkaConsumer<String, String> consumer = KafkaConsumerTest.getConsumer();

        // 主题订阅
        consumer.subscribe(KafkaCommonProperties.TOPIC);

        // 开启线程，进行监听
        while (Boolean.TRUE) {
            ConsumerRecords<String, String> records = consumer.poll(1000);

            records.forEach(record -> {
                logger.info(">>>>>>>>消费者进行消费 offset:{}, value:{}", record.offset(), record.value());
            });
        }
    }

}
